package gamma

import (
	"context"
	"fmt"
	"github.com/gogf/gf/v2/container/glist"
	"github.com/gogf/gf/v2/container/gqueue"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/panjf2000/ants/v2"
	"math/rand"
	"sync"
	"time"
)

const RetryLimitCount = 3

type TranslationRequestBucket struct {
	Content    string
	RetryLimit int
	RetryCount int
}

type ApiCallParam struct {
	ctx         context.Context
	bucketQueue *gqueue.Queue
	resultList  *glist.List // holds the request entity for simplifying testing
	wg          *sync.WaitGroup
	taskCh      chan *ApiCallParam
}

func CallOpenAiApi(input interface{}) {
	defer func() {
		fmt.Println("==================== LEAVE ====================")
	}()

	param := input.(*ApiCallParam)

	triggerNextAction := func() {
		go func() {
			fmt.Printf("SEND NEW TASK: start\n")
			param.taskCh <- param
			fmt.Printf("SEND NEW TASK: end\n")
		}()
	}

	item := param.bucketQueue.Pop()
	if item == nil {
		fmt.Printf("bucket queue is empty now !!!!!!!!!!!!!!!!!!\n")
		return
	}
	requestBucket := item.(*TranslationRequestBucket)

	mimicSeed := rand.Intn(4) // data range is [0, 3)
	//mimicSeed := rand.Intn(10) // data range is [0, 3)

	if 0 == mimicSeed {
		// mimic rate limit error
		if requestBucket.RetryCount < requestBucket.RetryLimit {
			requestBucket.RetryCount++
			param.wg.Add(1)

			g.Log().Infof(param.ctx, "rate limit")
			time.Sleep(time.Duration(50+rand.Intn(200)) * time.Millisecond)

			param.bucketQueue.Push(requestBucket)
			triggerNextAction()
		} else {
			g.Log().Warningf(param.ctx, "retry limitation reached")

			requestBucket.Content = "failed after retrying times"
			param.resultList.PushBack(requestBucket)
		}
		return
	}

	if 1 == mimicSeed {
		// mimic bucket truncated error
		g.Log().Infof(param.ctx, "translation truncated")

		param.wg.Add(2)

		param.bucketQueue.Push(&TranslationRequestBucket{
			Content:    fmt.Sprintf("%s-%d-1", requestBucket.Content, time.Now().UnixMilli()),
			RetryLimit: RetryLimitCount,
			RetryCount: 0,
		})

		param.bucketQueue.Push(&TranslationRequestBucket{
			Content:    fmt.Sprintf("%s-%d-2", requestBucket.Content, time.Now().UnixMilli()),
			RetryLimit: RetryLimitCount,
			RetryCount: 0,
		})

		triggerNextAction()
		triggerNextAction()

		return
	}

	if 2 == mimicSeed {
		// mimic other error

		if requestBucket.RetryCount < requestBucket.RetryLimit {
			requestBucket.RetryCount++

			param.wg.Add(1)
			param.bucketQueue.Push(requestBucket)

			triggerNextAction()
		} else {
			g.Log().Warningf(param.ctx, "retry limitation reached")

			requestBucket.Content = "failed after retrying times"
			param.resultList.PushBack(requestBucket)
		}

		return
	}

	// not mimic the translation missing error
	requestBucket.Content += " OK"
	param.resultList.PushBack(requestBucket)
	g.Log().Infof(param.ctx, "add on result into result list")
}

func Entrance(messageCount int) *glist.List {
	ctx, cancel := context.WithCancel(context.Background())

	taskCh := make(chan *ApiCallParam, messageCount)

	bucketQueue := gqueue.New()
	for i := 0; i < messageCount; i++ {
		bucketQueue.Push(&TranslationRequestBucket{
			Content:    fmt.Sprintf("message_%d", i),
			RetryLimit: RetryLimitCount,
			RetryCount: 0,
		})
	}

	responseBucketList := glist.New(true)
	poolOpenAiApi, _ := ants.NewPoolWithFunc(10, func(input interface{}) {
		CallOpenAiApi(input)
	})

	var wg sync.WaitGroup
	var stepWg sync.WaitGroup
	stepWg.Add(3)

	wg.Add(int(bucketQueue.Len()))
	go func() {
		defer stepWg.Done()

		i := 0
		for loop := true; loop; {
			select {
			case <-ctx.Done():
				loop = false
				continue
			default:
				if i >= int(bucketQueue.Len()) {
					loop = false
					continue
				}
				taskCh <- &ApiCallParam{
					ctx:         ctx,
					bucketQueue: bucketQueue,
					resultList:  responseBucketList,
					wg:          &wg,
					taskCh:      taskCh,
				}
				i++
			}
		}
	}()

	go func() {
		defer stepWg.Done()

		for loop := true; loop; {
			select {
			case <-ctx.Done():
				loop = false
				continue
			case param, ok := <-taskCh:
				if !ok {
					loop = false
					continue
				}

				fmt.Printf("[pool] processing request: start\n")
				CallOpenAiApi(param)
				fmt.Printf("[pool] processing request: finished\n")
				fmt.Printf("wg rel")
				param.wg.Done()

			}
		}
	}()

	go func() {
		defer stepWg.Done()

		wg.Wait()
		cancel()
	}()

	stepWg.Add(1)
	go func() {
		defer stepWg.Done()

		for loop := true; loop; {
			select {
			case <-ctx.Done():
				loop = false
				continue
			default:
				time.Sleep(200 * time.Millisecond)
				fmt.Printf("current request queue size : %d\n", bucketQueue.Len())
			}
		}
	}()

	stepWg.Wait()
	bucketQueue.Close()
	poolOpenAiApi.Release()

	return responseBucketList
}

/*
[pool] processing request: finished
[pool] processing request: start
2023-06-13 16:58:04.065 [INFO] add on result into result list
==================== LEAVE ====================
panic: sync: negative WaitGroup counter

goroutine 15 [running]:
sync.(*WaitGroup).Add(0x1400065cdd8?, 0x1?)
	/usr/local/go/src/sync/waitgroup.go:62 +0x10c
sync.(*WaitGroup).Done(0x105005d00?)
	/usr/local/go/src/sync/waitgroup.go:87 +0x20
example.com/prototype/internal/gamma.CallOpenAiApi({0x104ff47e0?, 0x1400009c1e0})
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:124 +0x400
example.com/prototype/internal/gamma.Entrance.func3()
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:186 +0x128
created by example.com/prototype/internal/gamma.Entrance
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:176 +0x398


Process finished with the exit code 1


~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~`
==================== LEAVE ====================
panic: sync: negative WaitGroup counter

goroutine 15 [running]:
sync.(*WaitGroup).Add(0x140000bfdd8?, 0x1?)
	/usr/local/go/src/sync/waitgroup.go:62 +0x10c
sync.(*WaitGroup).Done(0x1400064c5d0?)
	/usr/local/go/src/sync/waitgroup.go:87 +0x20
example.com/prototype/internal/gamma.CallOpenAiApi({0x1025ac7e0?, 0x140003fc990})
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:118 +0x344
example.com/prototype/internal/gamma.Entrance.func3()
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:187 +0x128
created by example.com/prototype/internal/gamma.Entrance
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:177 +0x398
panic: sync: WaitGroup is reused before previous Wait has returned

goroutine 16 [running]:
sync.(*WaitGroup).Wait(0x140003d4ed0)
	/usr/local/go/src/sync/waitgroup.go:118 +0xb0
example.com/prototype/internal/gamma.Entrance.func4()
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:196 +0x58
created by example.com/prototype/internal/gamma.Entrance
	/Users/0826new-1/tmp/go_playground/prototype/internal/gamma/gamma.go:193 +0x404


Process finished with the exit code 1


*/
